<?php

namespace Library\Rabbitmq;

use \PhpAmqpLib\Exception\AMQPTimeoutException;

/**
 * rabbitmq 发布订阅模型
 */
class publish extends Msg
{
    protected $consumerAck = false;
    protected $exchange = 'publish_exchange'; 
    protected $queue = 'publish_queue'; 
    
    public function __construct($queue = '',$exchange = '',$isConsumer=false,$consumerAck = false)
    {
        //默认交换机队列rabbitmq
        if($queue){
            $this->queue = $queue;
        }
        if($exchange){
            $this->exchange = $exchange;
        }
        $this->consumerAck = $consumerAck;
        //parent::__construct();
        $this->connect();
        if(!$isConsumer){
            $this->queue_set_publish();
        }else{
            $this->queue_set_consumer();
        }
    }
    
    public function queue_set_publish()
    {
        $this->channel->exchange_declare($this->exchange, 'fanout', false, false, false);
        $this->channel->queue_declare($this->queue, false, false, false, false);
        $this->channel->queue_bind($this->queue, $this->exchange);
    }

    public function queue_set_consumer()
    {
        $this->channel->exchange_declare($this->exchange, 'fanout', false, false, false);
        //广播写法 暂时不明白
        list($queue_name, ,) = $this->channel->queue_declare("", false, false, false, false);
        $this->channel->queue_bind($queue_name, $this->exchange);
    }

    /**
     * 生产者 
     * 参数只有$content
     */
    public function publish($content, $exchange='', $routeKey='', $head = [])
    {
        return parent::publish($content, $this->exchange, $this->queue, $head = []);
    }

    /**
     * 消费者
     */
    public function consumer($call_back,$consumerTag='')
    {
        try{
            $this->channel->basic_qos( null , 2, null );
            if($this->consumerAck){
                $this->channel->basic_consume($this->queue, $consumerTag, false, false, false, false, $call_back);
            }else{
                $this->channel->basic_consume($this->queue, $consumerTag, false, true, false, false, $call_back); //不需要确认回复
            }
            while (count($this->channel->callbacks)) {
                $this->channel->wait();
            }
        }catch(AMQPTimeoutException $e){
            throw $e;
        }
        
    }
}